﻿using AbstractMq;
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace RabbitMqHelp
{
    public class RabbitMQSubscriber : ISubscriber
    {
        private readonly IConnection _IConnection;
        private readonly ILogger<RabbitMQSubscriber> _logger;
        public RabbitMQSubscriber(IConnection connection,ILoggerFactory loggerFactory)
        {
            _IConnection = connection;
            _logger = loggerFactory.CreateLogger<RabbitMQSubscriber>();
        }
        public void Subscribe<T>(string channelName, Action<byte[]> callback) where T : class
        {
            var _Channel = _IConnection.CreateModel();
            var queueName = _Channel.QueueDeclare().QueueName;
            //声明交换机
            _Channel.ExchangeDeclare(exchange: this.GetType().Assembly.GetName().Name, type: "direct", durable: true, autoDelete: false, arguments: null);
            //声明队列
            _Channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
            var properties = _Channel.CreateBasicProperties();
            properties.Persistent = true;
            //将队列与交换机进行绑定
            _Channel.QueueBind(queue: queueName, exchange: this.GetType().Assembly.GetName().Name, routingKey: channelName);
            //声明为手动确认，每次只消费1条消息。
            _Channel.BasicQos(0, 1, false);
            //定义消费者
            var consumer = new EventingBasicConsumer(_Channel);
            //接收事件
            consumer.Received += (eventSender, args) =>
            {
                var message = args.Body.ToArray();//接收到的消息

                callback(message);
                //返回消息确认
                _Channel.BasicAck(args.DeliveryTag, true);
            };
            //开启监听
            _Channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
        }

        public void Subscribe(string channelName, string group, Action<byte[]> callback)
        {
            var _Channel = _IConnection.CreateModel();
            //声明交换机
            _Channel.ExchangeDeclare(exchange: this.GetType().Assembly.GetName().Name, type: "fanout", durable: true, autoDelete: false, arguments: null);
            //声明队列
            _Channel.QueueDeclare(queue: group, durable: true, exclusive: false, autoDelete: false, arguments: null);
            var properties = _Channel.CreateBasicProperties();
            properties.Persistent = true;
            //将队列与交换机进行绑定
            _Channel.QueueBind(queue: group, exchange: this.GetType().Assembly.GetName().Name, routingKey: channelName);
            //声明为手动确认，每次只消费1条消息。
            _Channel.BasicQos(0, 1, false);
            //定义消费者
            var consumer = new EventingBasicConsumer(_Channel);
            //接收事件
            consumer.Received += (eventSender, args) =>
            {
                var message = args.Body.ToArray();//接收到的消息

                callback(message);
                //返回消息确认
                _Channel.BasicAck(args.DeliveryTag, true);
            };
            //开启监听
            _Channel.BasicConsume(queue: group, autoAck: false, consumer: consumer);
        }

        public void Subscribe(string exchangeName, string routingKey, string queueName, Action<byte[]> callback)
        {
            var _Channel = _IConnection.CreateModel();
            //声明交换机
            _Channel.ExchangeDeclare(exchange: exchangeName, type: "fanout", durable: true, autoDelete: false, arguments: null);
            //声明队列
            _Channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
            var properties = _Channel.CreateBasicProperties();
            properties.Persistent = true;
            //将队列与交换机进行绑定
            _Channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routingKey);
            //声明为手动确认，每次只消费1条消息。
            _Channel.BasicQos(0, 1, false);
            //定义消费者
            var consumer = new EventingBasicConsumer(_Channel);
            //接收事件
            consumer.Received += (eventSender, args) =>
            {
                var message = args.Body.ToArray();//接收到的消息

                callback(message);
                //返回消息确认
                _Channel.BasicAck(args.DeliveryTag, true);
            };
            //开启监听
            _Channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
        }

        public void Subscribe(string exchangeName, string routingKey, string queueName, Func<byte[],bool> callback)
        {
            var _Channel = _IConnection.CreateModel();
            //声明交换机
            _Channel.ExchangeDeclare(exchange: exchangeName, type: "fanout", durable: true, autoDelete: false, arguments: null);
            //声明队列
            _Channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
            var properties = _Channel.CreateBasicProperties();
            properties.Persistent = true;
            //将队列与交换机进行绑定
            _Channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routingKey);
            //声明为手动确认，每次只消费1条消息。
            _Channel.BasicQos(0, 1, false);
            //定义消费者
            var consumer = new EventingBasicConsumer(_Channel);
            //接收事件
            consumer.Received += (eventSender, args) =>
            {
                var message = args.Body.ToArray();//接收到的消息

                if (callback(message))
                {
                    //返回消息确认
                    _Channel.BasicAck(args.DeliveryTag, false);
                }
                else
                {
                    _Channel.BasicReject(deliveryTag: args.DeliveryTag, requeue: true);
                }
            };
            //开启监听
            _Channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
        }

    }
}
